package com.mkx.multitier.asynctask;

import com.mkx.multitier.AbstractLayer;
import com.mkx.multitier.Monitor;
import com.mkx.multitier.ResourceUsage;
import com.mkx.multitier.Task;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * 异步任务执行层（L4）组件，负责管理和执行异步任务
 */
public class AsyncTaskExecutor extends AbstractLayer {
    private final ScheduledExecutorService scheduledExecutor;
    private final ThreadPoolExecutor taskThreadPool;
    private final Map<String, BlockingQueue<AsyncTask>> taskQueues;
    private final AtomicInteger activeTasksCount = new AtomicInteger(0);
    private final AtomicLong taskScheduledCount = new AtomicLong(0);
    private final ResourceUsage resourceUsage = new ResourceUsage();
    private final AtomicReference<AsyncTask> highPriorityTaskSignal = new AtomicReference<>();
    private final CopyOnWriteArrayList<Runnable> taskProcessors = new CopyOnWriteArrayList<>();
    private volatile boolean running = false;
    private Thread monitorThread;

    // 性能监控参数
    private final long QUEUE_CHECK_INTERVAL_MS = 50; // 队列检查间隔
    private final int MAX_PROCESSORS = Math.max(2, Runtime.getRuntime().availableProcessors()); // 最大处理线程数
    private final int CORE_THREADS = Runtime.getRuntime().availableProcessors(); // 核心线程数
    private final int MAX_THREADS = Runtime.getRuntime().availableProcessors() * 8; // 增加最大线程数以适应IO密集型任务
    private final long KEEP_ALIVE_TIME = 60L; // 线程空闲存活时间
    private final int TASK_QUEUE_CAPACITY = 2000; // 增加任务队列容量

    public AsyncTaskExecutor(Monitor monitor) {
        super("L4-AsyncTaskExecutor", monitor);
        // 定时任务线程池，用于调度延迟执行的任务 - 使用ScheduledThreadPoolExecutor构造函数以更好地控制参数
        this.scheduledExecutor = new ScheduledThreadPoolExecutor(
                2, // 核心线程数
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "scheduled-task-thread-" + threadNumber.getAndIncrement());
                        thread.setDaemon(true);
                        return thread;
                    }
                },
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略：抛出异常
        );
        // 配置线程池参数
        ((ScheduledThreadPoolExecutor) this.scheduledExecutor).setKeepAliveTime(60L, TimeUnit.SECONDS);
        ((ScheduledThreadPoolExecutor) this.scheduledExecutor).allowCoreThreadTimeOut(false);
        // 异步任务执行线程池 - 优化线程池配置
        this.taskThreadPool = new ThreadPoolExecutor(
                CORE_THREADS, // 核心线程数
                MAX_THREADS, // 增加最大线程数以适应IO密集型任务
                KEEP_ALIVE_TIME, TimeUnit.SECONDS, // 线程空闲超时时间
                new LinkedBlockingQueue<>(TASK_QUEUE_CAPACITY), // 增加队列容量，避免过早触发拒绝策略
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "async-task-thread-" + threadNumber.getAndIncrement());
                        thread.setDaemon(true);
                        thread.setPriority(Thread.NORM_PRIORITY); // 设置适当的线程优先级
                        return thread;
                    }
                },
                new OptimizedRejectedExecutionHandler() // 优化的拒绝策略
        );

        // 允许核心线程超时，当系统负载较低时释放资源
        this.taskThreadPool.allowCoreThreadTimeOut(true);

        // 设置线程池的饱和策略监控
        ((ThreadPoolExecutor) this.taskThreadPool).setRejectedExecutionHandler(new OptimizedRejectedExecutionHandler());

        this.taskQueues = new ConcurrentHashMap<>();
    }

    /**
     * 报告资源使用情况
     */
    public void reportResourceUsage(ResourceUsage usage) {
        // 直接输出资源使用情况
        System.out.println("Resource usage report: " + usage);
    }

    /**
     * 获取层名称
     */
    @Override
    public String getLayerName() {
        return "L4-AsyncTaskExecutor";
    }

    @Override
    protected void doInitialize() {
        System.out.println("Initializing L4 AsyncTaskExecutor...");
        // 初始化任务队列和线程池
    }

    @Override
    protected void doStart() {
        System.out.println("Starting L4 AsyncTaskExecutor...");
        running = true;

        // 启动多个任务处理线程，根据CPU核心数确定
        int processorCount = Math.min(MAX_PROCESSORS, Runtime.getRuntime().availableProcessors());
        for (int i = 0; i < processorCount; i++) {
            Runnable processor = this::processTaskQueues;
            taskProcessors.add(processor);
            scheduledExecutor.submit(processor);
        }

        // 启动监控线程
        monitorThread = new Thread(this::monitorResourceUsage);
        monitorThread.setDaemon(true);
        monitorThread.start();

        // 启动动态线程池调整任务
        scheduledExecutor.scheduleWithFixedDelay(this::adjustThreadPoolSize, 30, 30, TimeUnit.SECONDS);
    }

    @Override
    protected void doShutdown() {
        System.out.println("Shutting down L4 AsyncTaskExecutor...");
        running = false;

        // 清空任务信号
        highPriorityTaskSignal.set(null);

        // 关闭线程池
        scheduledExecutor.shutdown();
        taskThreadPool.shutdown();

        try {
            // 给予更多时间让任务完成
            boolean scheduledTerminated = scheduledExecutor.awaitTermination(10, TimeUnit.SECONDS);
            boolean taskTerminated = taskThreadPool.awaitTermination(10, TimeUnit.SECONDS);

            if (!scheduledTerminated || !taskTerminated) {
                // 强制关闭
                scheduledExecutor.shutdownNow();
                taskThreadPool.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // 中断后仍然尝试强制关闭
            scheduledExecutor.shutdownNow();
            taskThreadPool.shutdownNow();
        }

        // 中断监控线程
        if (monitorThread != null) {
            monitorThread.interrupt();
        }
    }

    @Override
    protected void doProcessTask(Task task) {
        if (task instanceof AsyncTask) {
            scheduleTask((AsyncTask) task);
        } else {
            task.execute();
        }
    }

    /**
     * 调度异步任务
     */
    public void scheduleTask(AsyncTask task) {
        if (!running) {
            throw new IllegalStateException("AsyncTaskExecutor is not running");
        }

        taskScheduledCount.incrementAndGet();

        // 计算延迟时间
        Duration delay = null;
        if (task.getExecuteTime() != null) {
            delay = Duration.between(Instant.now(), task.getExecuteTime());
        } else if (task instanceof AbstractAsyncTask && ((AbstractAsyncTask) task).getDelay() != null) {
            delay = ((AbstractAsyncTask) task).getDelay();
        }

        if (delay != null && !delay.isNegative() && !delay.isZero()) {
            // 延迟执行任务
            scheduledExecutor.schedule(() -> {
                try {
                    executeTask(task);
                } catch (Exception e) {
                    handleTaskFailure(task, e);
                }
            }, delay.toMillis(), TimeUnit.MILLISECONDS);
        } else {
            // 立即执行任务
            executeTask(task);
        }
    }

    /**
     * 执行异步任务
     */
    private void executeTask(AsyncTask task) {
        // 获取或创建特定任务类型的优先队列
        String taskType = task.getTaskType();
        // 使用PriorityBlockingQueue替代LinkedBlockingQueue，实现基于优先级的任务处理
        BlockingQueue<AsyncTask> queue = taskQueues.computeIfAbsent(taskType, k ->
                new PriorityBlockingQueue<>(1000, Comparator.comparingInt(AsyncTask::getPriority)));

        try {
            // 优化：对于高负载情况，使用offer替代put，避免线程阻塞
            boolean added = queue.offer(task, 500, TimeUnit.MILLISECONDS);
            if (added) {
                resourceUsage.incrementQueueSize();

                // 对于高优先级任务，通过AtomicReference信号立即唤醒处理线程
                if (task.getPriority() <= 3) { // 优化优先级定义，1-3为高优先级
                    highPriorityTaskSignal.set(task);
                }
            } else {
                // 如果队列已满，尝试直接提交到线程池
                try {
                    submitTaskDirectly(task);
                } catch (RejectedExecutionException e) {
                    // 如果线程池也饱和，处理失败
                    handleTaskFailure(task, new RejectedExecutionException("Both task queue and thread pool are saturated"));
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            handleTaskFailure(task, e);
        }
    }

    /**
     * 直接将任务提交到线程池执行
     */
    private void submitTaskDirectly(AsyncTask task) {
        if (!running) return;

        activeTasksCount.incrementAndGet();
        resourceUsage.incrementActiveTasks();

        taskThreadPool.submit(() -> {
            long startTime = System.currentTimeMillis();
            try {
                task.execute();
                resourceUsage.incrementCompletedTasks();

                if (monitor != null) {
                    monitor.onTaskProcessed(getLayerName(), task, System.currentTimeMillis() - startTime);
                }
            } catch (Exception e) {
                handleTaskFailure(task, e);
            } finally {
                activeTasksCount.decrementAndGet();
                resourceUsage.decrementActiveTasks();
            }
        });
    }

    /**
     * 处理任务队列中的任务 - 优化版本
     */
    private void processTaskQueues() {
        final long threadId = Thread.currentThread().getId();

        while (running) {
            try {
                AsyncTask task = null;

                // 快速路径：检查是否有高优先级任务信号
                if (highPriorityTaskSignal.getAndSet(null) != null) {
                    // 有高优先级任务，立即查找并处理
                    task = findHighestPriorityTask();
                }

                // 如果没有找到高优先级任务或没有信号，则尝试获取任何可用任务
                if (task == null) {
                    task = findNextTaskToProcess();
                }

                if (task != null) {
                    // 任务不为空，提交到线程池执行
                    submitTaskForExecution(task);
                } else {
                    // 没有任务，使用非阻塞的短暂休眠，避免CPU空转
                    Thread.sleep(QUEUE_CHECK_INTERVAL_MS);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                System.err.println("Error in task processor thread-" + threadId + ": " + e.getMessage());
                // 异常处理，避免处理线程意外退出
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    /**
     * 查找所有队列中优先级最高的任务
     */
    private AsyncTask findHighestPriorityTask() {
        AsyncTask highestPriorityTask = null;
        BlockingQueue<AsyncTask> targetQueue = null;

        // 优化遍历：使用任务类型的快照，避免遍历过程中的并发修改问题
        List<String> taskTypes = new ArrayList<>(taskQueues.keySet());
        for (String taskType : taskTypes) {
            BlockingQueue<AsyncTask> queue = taskQueues.get(taskType);
            if (queue != null && !queue.isEmpty()) {
                AsyncTask task = queue.peek();
                if (task != null) {
                    // 优先级数字越小，优先级越高
                    if (highestPriorityTask == null || task.getPriority() < highestPriorityTask.getPriority()) {
                        highestPriorityTask = task;
                        targetQueue = queue;
                    }
                }
            }
        }

        // 找到任务后从队列中移除
        if (highestPriorityTask != null && targetQueue != null) {
            return targetQueue.poll();
        }
        return null;
    }

    /**
     * 查找下一个要处理的任务，优先考虑非空队列
     */
    private AsyncTask findNextTaskToProcess() {
        // 优化：先检查是否有队列包含任务
        List<BlockingQueue<AsyncTask>> nonEmptyQueues = taskQueues.values().stream()
                .filter(queue -> !queue.isEmpty())
                .collect(Collectors.toList());

        if (nonEmptyQueues.isEmpty()) {
            return null;
        }

        // 从非空队列中随机选择一个，避免资源竞争
        int randomIndex = ThreadLocalRandom.current().nextInt(nonEmptyQueues.size());
        BlockingQueue<AsyncTask> selectedQueue = nonEmptyQueues.get(randomIndex);

        return selectedQueue.poll();
    }

    /**
     * 提交任务到线程池执行
     */
    private void submitTaskForExecution(AsyncTask task) {
        if (task == null) return;

        resourceUsage.decrementQueueSize();
        activeTasksCount.incrementAndGet();
        resourceUsage.incrementActiveTasks();

        try {
            taskThreadPool.submit(() -> {
                long startTime = System.currentTimeMillis();
                try {
                    task.execute();
                    resourceUsage.incrementCompletedTasks();

                    if (monitor != null) {
                        monitor.onTaskProcessed(getLayerName(), task, System.currentTimeMillis() - startTime);
                    }
                } catch (Exception e) {
                    handleTaskFailure(task, e);
                } finally {
                    activeTasksCount.decrementAndGet();
                    resourceUsage.decrementActiveTasks();
                }
            });
        } catch (RejectedExecutionException e) {
            // 如果线程池拒绝任务，回退到处理失败逻辑
            activeTasksCount.decrementAndGet();
            resourceUsage.decrementActiveTasks();
            handleTaskFailure(task, e);
        }
    }

    /**
     * 处理任务失败
     */
    private void handleTaskFailure(AsyncTask task, Exception e) {
        resourceUsage.incrementFailedTasks();

        System.err.println("Task failed: " + task + ", error: " + e.getMessage());

        // 处理重试逻辑
        if (task.isRetryable() && task.getCurrentRetryCount() < task.getMaxRetries()) {
            task.incrementRetryCount();
            long backoffTime = calculateBackoffTime(task.getCurrentRetryCount());

            System.out.println("Retrying task " + task +
                    " (attempt " + task.getCurrentRetryCount() + "/" + task.getMaxRetries() +
                    ", backoff: " + backoffTime + "ms)");

            scheduledExecutor.schedule(() -> scheduleTask(task), backoffTime, TimeUnit.MILLISECONDS);
        } else {
            System.err.println("Task " + task + " failed permanently after " +
                    task.getMaxRetries() + " retries");
        }
    }

    /**
     * 计算退避时间（指数退避）
     */
    private long calculateBackoffTime(int retryCount) {
        // 基础退避时间 100ms，最大 30s，添加随机抖动
        long baseBackoff = 100L;
        long maxBackoff = 30000L;
        long backoff = Math.min(baseBackoff * (1L << retryCount), maxBackoff);
        // 添加随机抖动（±10%）
        return (long) (backoff * (0.9 + Math.random() * 0.2));
    }

    /**
     * 监控资源使用情况 - 优化版本
     */
    private void monitorResourceUsage() {
        while (running) {
            try {
                // 更频繁地收集数据，但减少报告频率
                Thread.sleep(2000); // 每2秒收集一次数据

                // 收集详细的线程池状态
                int activeThreads = taskThreadPool.getActiveCount();
                int poolSize = taskThreadPool.getPoolSize();
                int queueSize = taskThreadPool.getQueue().size();
                long completedTasks = taskThreadPool.getCompletedTaskCount();

                // 更新资源使用情况
                resourceUsage.setActiveTasks(activeTasksCount.get());
                resourceUsage.setThreadCount(activeThreads);

                // 计算任务队列总大小
                long totalQueueSize = taskQueues.values().stream()
                        .mapToLong(BlockingQueue::size)
                        .sum();
                resourceUsage.setQueueSize(totalQueueSize);

                // 注意：ResourceUsage类不支持这些额外的监控指标设置方法
                // 已使用现有的setCompletedTasks方法（通过increment方式）
                // 线程池大小和任务队列大小信息可以通过日志输出

                // 直接报告资源使用情况
                reportResourceUsage(resourceUsage);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    /**
     * 动态调整线程池大小
     */
    private void adjustThreadPoolSize() {
        if (!running) return;

        try {
            int currentActive = taskThreadPool.getActiveCount();
            int currentPoolSize = taskThreadPool.getPoolSize();
            int queueSize = taskThreadPool.getQueue().size();

            // 根据队列大小和活跃线程数动态调整核心线程数
            if (queueSize > currentActive * 2 && currentPoolSize < MAX_THREADS) {
                // 队列积压，增加线程数
                int newPoolSize = Math.min(currentPoolSize * 2, MAX_THREADS);
                taskThreadPool.setMaximumPoolSize(newPoolSize);
                System.out.println("Increasing thread pool size to " + newPoolSize + " due to queue backlog");
            } else if (queueSize < currentActive && currentPoolSize > CORE_THREADS) {
                // 线程过剩，减少线程数
                int newPoolSize = Math.max(currentPoolSize / 2, CORE_THREADS);
                taskThreadPool.setMaximumPoolSize(newPoolSize);
                System.out.println("Decreasing thread pool size to " + newPoolSize + " due to low queue load");
            }
        } catch (Exception e) {
            System.err.println("Error adjusting thread pool size: " + e.getMessage());
        }
    }

    /**
     * 优化的拒绝执行处理器
     */
    private class OptimizedRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 统计拒绝次数 - 使用现有的incrementFailedTasks方法作为替代
            resourceUsage.incrementFailedTasks();
            System.out.println("Task rejected, incrementing failed tasks count");

            // 尝试使用调用者线程执行任务
            if (!executor.isShutdown()) {
                try {
                    // 使用调用者线程执行，但设置超时
                    ThreadPoolExecutor.CallerRunsPolicy callerRuns = new ThreadPoolExecutor.CallerRunsPolicy();
                    callerRuns.rejectedExecution(r, executor);
                } catch (Exception e) {
                    System.err.println("Task rejection handler failed: " + e.getMessage());
                }
            }
        }
    }

    /**
     * 抽象异步任务基类，提供AsyncTask接口的基本实现
     */
    public abstract static class AbstractAsyncTask implements AsyncTask, Comparable<AbstractAsyncTask> {
        private final String taskId;
        private final String taskType;
        private final int priority;
        private final boolean retryable;
        private final int maxRetries;
        private final AtomicInteger currentRetryCount = new AtomicInteger(0);
        private Instant executeTime;
        private Duration delay;

        public AbstractAsyncTask(String taskType, int priority, boolean retryable, int maxRetries) {
            this.taskId = java.util.UUID.randomUUID().toString();
            this.taskType = taskType;
            this.priority = priority;
            this.retryable = retryable;
            this.maxRetries = maxRetries;
        }

        @Override
        public String getTaskId() {
            return taskId;
        }

        @Override
        public int getPriority() {
            return priority;
        }

        @Override
        public String getTaskType() {
            return taskType;
        }

        @Override
        public boolean isRetryable() {
            return retryable;
        }

        @Override
        public int getMaxRetries() {
            return maxRetries;
        }

        @Override
        public int getCurrentRetryCount() {
            return currentRetryCount.get();
        }

        @Override
        public void incrementRetryCount() {
            currentRetryCount.incrementAndGet();
        }

        @Override
        public Instant getExecuteTime() {
            return executeTime;
        }

        public void setExecuteTime(Instant executeTime) {
            this.executeTime = executeTime;
        }

        @Override
        public void setDelay(Duration delay) {
            this.delay = delay;
        }

        public Duration getDelay() {
            return delay;
        }

        @Override
        public String toString() {
            return "AsyncTask{" +
                    "taskId='" + taskId + "'" +
                    ", taskType='" + taskType + "'" +
                    ", priority=" + priority +
                    ", retryable=" + retryable +
                    ", maxRetries=" + maxRetries +
                    ", currentRetries=" + currentRetryCount.get() +
                    ", executeTime=" + (executeTime != null ? executeTime : "immediate") +
                    '}';
        }

        /**
         * 实现Comparable接口，便于任务比较和排序
         */
        @Override
        public int compareTo(AbstractAsyncTask other) {
            // 首先按优先级排序
            int priorityCompare = Integer.compare(this.priority, other.priority);
            if (priorityCompare != 0) {
                return priorityCompare;
            }

            // 优先级相同时，按执行时间排序
            if (this.executeTime != null && other.executeTime != null) {
                return this.executeTime.compareTo(other.executeTime);
            } else if (this.executeTime == null && other.executeTime != null) {
                return -1; // 无执行时间的优先于有执行时间的
            } else if (this.executeTime != null && other.executeTime == null) {
                return 1;
            }

            // 其他情况按任务ID排序，确保排序稳定性
            return this.taskId.compareTo(other.taskId);
        }
    }
}
